package com.clp.util;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

public class KafkaUtil {
    /**
     * 检查指定服务器是否有列表中的所有主题，没有就创建
     *
     * @param bootStrapServers：kafka服务器
     * @param topicList：主题列表
     */
    public static void createTopicsIfNotExist(String bootStrapServers, List<String> topicList) {
        Properties prop = new Properties();
        prop.put("bootstrap.servers", bootStrapServers);

        AdminClient adminClient = AdminClient.create(prop);
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        KafkaFuture<Set<String>> future = listTopicsResult.names();

        try {
            // 检查是否有需要的主题，没有就创建
            ArrayList<NewTopic> newTopicList = new ArrayList<NewTopic>();
            Set<String> topics = future.get();

            // 检查所需要的主题是否都已经创建好
            for (String topicStr : topicList) {
                if (!topics.contains(topicStr)) {
                    // 创建主题。  参数：主题名称、分区数、副本数
                    newTopicList.add(new NewTopic(topicStr, 1, (short) 1));
                }
            }
            // 创建未创建的主题
            if (!newTopicList.isEmpty()) {
                CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopicList);
                createTopicsResult.all().get();
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } finally {
            adminClient.close();
        }
    }
}
